package com.rem.flink.flink5Watermark;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * 增量规约函数
 * ReduceFunction 规约函数 窗口中收集到的数据两两进行归约 输入输出类型必须一致
 *
 * @author Rem
 * @date 2022-10-11
 */

public class WindowReduceTest {

    /**
     * 求出 name 出现的频率
     *
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.从socket中读取流
        DataStreamSource<Event> stream = streamEnv.addSource(new ClickSource());

        // 插入水位线的逻辑
        stream.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()))
                .map((MapFunction<Event, Tuple2<String, Long>>) value -> Tuple2.of(value.getUser(), 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG))
                .keyBy(a -> a.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))

                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2)  {
                        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                    }
                })
                .print();

        //7.执行流
        streamEnv.execute();

    }
}
